package com.youxuit.youxu.kecheng.CASandAtomic;


import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * DataPush
 * <p>
 * 消息消费者
 *
 * @author ZhenXu
 * @date 2023-12-20 15:13
 **/
public class MessageConsumer extends Thread {

    /**
     * 线程运行标识，可通过该标识来控制消费者线程启停
     */
    public volatile boolean threadFlag = true;

    /**
     * 待消费消息队列，此处将其长度设置为1000W
     */

    private static final LinkedBlockingQueue<Integer> MESSAGE_BLOCKING_QUEUE = new LinkedBlockingQueue<Integer>(10000000);

    /**
     * 说明: 推送消息到消息队列中
     *
     * @param message 消息实体
     * @throws InterruptedException 可针对异常情况进行推送补偿
     * @author ZhenXu
     * @date 2023-12-21 21:00
     */
    public static void pushMsg(Integer message) throws InterruptedException {
        MESSAGE_BLOCKING_QUEUE.put(message);
    }

    /**
     * 说明: 从阻塞队列中获取消息
     *
     * @return {@link Integer}
     * @throws InterruptedException 可针对异常情况进行消费补偿
     * @author ZhenXu
     * @date 2023-12-21 21:20
     */
    public static Integer getMsg() throws InterruptedException {
        // 如果阻塞队列中没有消息则会阻塞
        return MESSAGE_BLOCKING_QUEUE.take();
    }

    /**
     * 说明: 从阻塞队列中获取消息，如果队列为空则会等待60分钟
     * <p>
     * 有元素则立马返回队列中的头部元素，若没有则等待60分钟，超时后则唤醒线程
     *
     * @return {@link Integer}
     * @throws InterruptedException 可针对异常情况进行消费补偿
     * @author ZhenXu
     * @date 2023-12-21 21:54
     */
    public static Integer poll() throws InterruptedException {
        return MESSAGE_BLOCKING_QUEUE.poll(60, TimeUnit.MINUTES);
    }

    @Override
    public void run() {
        Integer message = null;
        System.out.println("消息消费线程启动...");
        while (this.threadFlag) {
            try {
                // 获取消息，查看阻塞队列的长度，打印会影响消费效率
//                System.out.println(MESSAGE_BLOCKING_QUEUE.size());

                //消费这里有没有增加效率的方式呢？暂时只想到了增加消费者，但是单个消费者该如何优化没有想到。
                message = MessageConsumer.getMsg();

                //输出消息，打印会影响消费效率
//                System.out.println("消费消息：" + message);
            } catch (Exception e) {
                //若此处消费消息出现异常，则可以进行消费补偿动作，重新消费或者根据业务场景而定。
                System.out.println("消费消息异常");
            }
        }
        System.out.println("消费消息线程停止");
    }
}